package io.kiki.stack.netty.processor;


import io.kiki.stack.netty.channelManager.ChannelContext;
import io.kiki.stack.netty.command.AbstractRequestCommand;
import io.kiki.stack.netty.command.Command;
import io.kiki.stack.netty.command.CommandFactory;
import io.kiki.stack.netty.command.ResponseType;
import io.kiki.stack.netty.exception.DeserializationException;
import io.kiki.stack.netty.exception.SerializationException;
import io.kiki.stack.netty.invoker.RequestContext;
import io.kiki.stack.netty.invoker.RpcCommandType;
import io.kiki.stack.netty.protocol.DeserializeLevel;
import io.kiki.stack.netty.userProcessor.AsyncMultiInterestUserProcessor;
import io.kiki.stack.netty.userProcessor.AsyncUserProcessor;
import io.kiki.stack.netty.userProcessor.UserProcessor;
import io.kiki.stack.netty.util.InetSocketAddressUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;


@Slf4j
@Setter
@Getter
public class RequestProcessor extends AbstractProcessor<AbstractRequestCommand> {


    public RequestProcessor(CommandFactory commandFactory) {
        super(commandFactory);
    }


    @Override
    public void process(ChannelContext channelContext, AbstractRequestCommand requestCommand, Executor defaultExecutor) throws Exception {
        if (!deserializeRequestCommand(channelContext, requestCommand, DeserializeLevel.deserialize_clazz)) {
            return;
        }
        UserProcessor userProcessor = channelContext.getUserProcessor(requestCommand.getRequestClass());
        if (userProcessor == null) {
            String errMsg = "No user processor found for request: " + requestCommand.getRequestClass();
            log.error(errMsg);
            sendResponseIfNecessary(channelContext, requestCommand.getType(), this.getCommandFactory().createExceptionResponse(requestCommand.getId(), errMsg));
            return;// must end process
        }

        // set timeout check state from user's processor
        channelContext.setTimeoutDiscard(userProcessor.timeoutDiscard());

        // to check whether to process in io thread
        if (userProcessor.processInIOThread()) {
            if (!deserializeRequestCommand(channelContext, requestCommand, DeserializeLevel.deserialize_all)) {
                return;
            }
            // process in io thread
            new RequestProcessTask(this, channelContext, requestCommand).run();
            return;// end
        }

        Executor executor;
        // to check whether get executor using executor selector
        if (null == userProcessor.getExecutorSelector()) {
            executor = userProcessor.getExecutor();
        } else {
            // in case haven't deserialized in io thread
            // it need to deserialize clazz and header before using executor dispath strategy
            if (!deserializeRequestCommand(channelContext, requestCommand, DeserializeLevel.deserialize_header)) {
                return;
            }
            //try get executor with strategy
            executor = userProcessor.getExecutorSelector().select(requestCommand.getRequestClass(), requestCommand.getRequestHeader());
        }

        // Till now, if executor still null, then try default
        if (executor == null) {
            executor = (this.getExecutor() == null ? defaultExecutor : this.getExecutor());
        }

        requestCommand.setBeforeEnterQueueTime(System.nanoTime());
        // use the final executor dispatch process task
        executor.execute(new RequestProcessTask(this, channelContext, requestCommand));
    }


    @Override
    public void doProcess(final ChannelContext channelContext, AbstractRequestCommand requestCommand) throws Exception {
        long currentTimestamp = System.currentTimeMillis();


        preProcessRemotingContext(channelContext, requestCommand, currentTimestamp);
        if (channelContext.isTimeoutDiscard() && channelContext.isRequestTimeout()) {
            timeoutLog(requestCommand, currentTimestamp, channelContext);// do some log
            return;// then, discard this request
        }
        //debugLog(ctx, cmd, currentTimestamp);
        // decode request all
        if (!deserializeRequestCommand(channelContext, requestCommand, DeserializeLevel.deserialize_all)) {
            return;
        }
        dispatchToUserProcessor(channelContext, requestCommand);
    }

    /**
     * Send response using remoting context if necessary.<br>
     * If request type is oneway, no need to send any response nor exception.
     */
    public void sendResponseIfNecessary(final ChannelContext channelContext, byte type, Command serializedResponse) {
        final int id = serializedResponse.getId();
        if (type != RpcCommandType.REQUEST_ONEWAY) {
            //Command serializedResponse = response;
            try {
                serializedResponse.serialize();
            } catch (SerializationException e) {
                String errMsg = "SerializationException occurred when sendResponseIfNecessary in RpcRequestProcessor, id=" + id;
                log.error(errMsg, e);
                serializedResponse = this.getCommandFactory().createExceptionResponse(id, ResponseType.server_serial_exception, e);
                try {
                    serializedResponse.serialize();// serialize again for exception response
                } catch (SerializationException e1) {
                    // should not happen
                    log.error("serialize SerializationException response failed!");
                }
            } catch (Throwable t) {
                String errMsg = "Serialize RpcResponseCommand failed when sendResponseIfNecessary in RpcRequestProcessor, id=" + id;
                log.error(errMsg, t);
                serializedResponse = this.getCommandFactory().createExceptionResponse(id, errMsg, t);
                try {
                    serializedResponse.serialize();// serialize again for exception response
                } catch (Throwable t1) {
                    // should not happen
                    log.error("serialize exception response failed!", t1);
                }
            }

            channelContext.writeAndFlush(serializedResponse).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (log.isDebugEnabled()) {
                        log.debug("Rpc response sent! requestId=" + id + ". The address is " + InetSocketAddressUtil.parseRemoteAddress(channelContext.getChannelHandlerContext().channel()));
                    }
                    if (!channelFuture.isSuccess()) {
                        log.error("Rpc response send failed! id=" + id + ". The address is " + InetSocketAddressUtil.parseRemoteAddress(channelContext.getChannelHandlerContext().channel()), channelFuture.cause());
                    }
                }
            });
        } else {
            if (log.isDebugEnabled()) {
                log.debug("Oneway rpc request received, do not send response, id=" + id + ", the address is " + InetSocketAddressUtil.parseRemoteAddress(channelContext.getChannelHandlerContext().channel()));
            }
        }
    }

    /**
     * dispatch request command to user processor
     */
    private void dispatchToUserProcessor(ChannelContext channelContext, AbstractRequestCommand requestCommand) {
        final int id = requestCommand.getId();
        final byte type = requestCommand.getType();
        // processor here must not be null, for it have been checked before
        UserProcessor processor = channelContext.getUserProcessor(requestCommand.getRequestClass());

        ClassLoader classLoader = null;
        try {
            ClassLoader bizClassLoader = processor.getBizClassLoader();
            if (bizClassLoader != null) {
                classLoader = Thread.currentThread().getContextClassLoader();
                Thread.currentThread().setContextClassLoader(bizClassLoader);
            }

            if (processor instanceof AsyncUserProcessor || processor instanceof AsyncMultiInterestUserProcessor) {
                try {
                    processor.asyncHandleRequest(processor.preHandleRequest(channelContext, requestCommand.getRequestObject()), new AsyncContextImpl(channelContext, requestCommand, this), requestCommand.getRequestObject());
                } catch (RejectedExecutionException e) {
                    log.warn("RejectedExecutionException occurred when do ASYNC process in RpcRequestProcessor");
                    sendResponseIfNecessary(channelContext, type, this.getCommandFactory().createExceptionResponse(id, ResponseType.server_threadpool_busy));
                } catch (Throwable t) {
                    String errMsg = "AYSNC process rpc request failed in RpcRequestProcessor, id=" + id;
                    log.error(errMsg, t);
                    sendResponseIfNecessary(channelContext, type, this.getCommandFactory().createExceptionResponse(id, errMsg, t));
                }
            } else {
                try {
                    Object responseObject = processor.syncHandleRequest(processor.preHandleRequest(channelContext, requestCommand.getRequestObject()), requestCommand.getRequestObject());

                    sendResponseIfNecessary(channelContext, type, this.getCommandFactory().createResponse(responseObject, requestCommand));
                } catch (RejectedExecutionException e) {
                    log.warn("RejectedExecutionException occurred when do SYNC process in RpcRequestProcessor");
                    sendResponseIfNecessary(channelContext, type, this.getCommandFactory().createExceptionResponse(id, ResponseType.server_threadpool_busy));
                } catch (Throwable t) {
                    String errMsg = "SYNC process rpc request failed in RpcRequestProcessor, id=" + id;
                    log.error(errMsg, t);
                    sendResponseIfNecessary(channelContext, type, this.getCommandFactory().createExceptionResponse(id, errMsg, t));
                }
            }
        } finally {
            if (classLoader != null) {
                Thread.currentThread().setContextClassLoader(classLoader);
            }
        }
    }

    /**
     * deserialize request command
     *
     * @return true if deserialize success; false if exception catched
     */
    private boolean deserializeRequestCommand(ChannelContext channelContext, AbstractRequestCommand cmd, int level) {
        boolean result;
        try {
            cmd.deserialize(level);
            result = true;
        } catch (DeserializationException e) {
            log.error("DeserializationException occurred when process in RpcRequestProcessor, id={}, deserializeLevel={}", cmd.getId(), DeserializeLevel.valueOf(level), e);
            sendResponseIfNecessary(channelContext, cmd.getType(), this.getCommandFactory().createExceptionResponse(cmd.getId(), ResponseType.server_deserial_exception, e));
            result = false;
        } catch (Throwable t) {
            String errMsg = "Deserialize RpcRequestCommand failed in RpcRequestProcessor, id=" + cmd.getId() + ", deserializeLevel=" + level;
            log.error(errMsg, t);
            sendResponseIfNecessary(channelContext, cmd.getType(), this.getCommandFactory().createExceptionResponse(cmd.getId(), errMsg, t));
            result = false;
        }
        return result;
    }

    /**
     * pre process remoting context, initial some useful infos and pass to biz
     *
     * @param channelContext   remoting context
     * @param requestCommand   rpc request command
     * @param currentTimestamp current timestamp
     */
    private void preProcessRemotingContext(ChannelContext channelContext, AbstractRequestCommand requestCommand, long currentTimestamp) {
        channelContext.setArriveTimestamp(requestCommand.getArriveTime());
        channelContext.setTimeout(requestCommand.getTimeout());
        channelContext.setRpcCommandType(requestCommand.getType());
        channelContext.getRequestContext().putIfAbsent(RequestContext._PROCESS_WAIT_TIME, currentTimestamp - requestCommand.getArriveTime());
        channelContext.getRequestContext().putIfAbsent(RequestContext._PROCESS_ARRIVE_HEADER_IN_NANO, requestCommand.getArriveHeaderTimeInNano());
        channelContext.getRequestContext().putIfAbsent(RequestContext._PROCESS_ARRIVE_BODY_IN_NANO, requestCommand.getArriveBodyTimeInNano());
        channelContext.getRequestContext().putIfAbsent(RequestContext._PROCESS_BEFORE_DISPATCH_IN_NANO, requestCommand.getBeforeEnterQueueTime());
        channelContext.getRequestContext().putIfAbsent(RequestContext._PROCESS_START_PROCESS_IN_NANO, System.nanoTime());
    }

    /**
     * print some log when request timeout and discarded in io thread.
     */
    private void timeoutLog(final AbstractRequestCommand requestCommand, long currentTimestamp, ChannelContext channelHandlerContext1) {
        if (log.isDebugEnabled()) {
            log.debug("request id [{}] currenTimestamp [{}] - arriveTime [{}] = server cost [{}] >= timeout value [{}].", requestCommand.getId(), currentTimestamp, requestCommand.getArriveTime(), (currentTimestamp - requestCommand.getArriveTime()), requestCommand.getTimeout());
        }

        String remoteAddr = "UNKNOWN";
        if (null != channelHandlerContext1) {
            ChannelHandlerContext channelHandlerContext = channelHandlerContext1.getChannelHandlerContext();
            Channel channel = channelHandlerContext.channel();
            if (null != channel) {
                remoteAddr = InetSocketAddressUtil.parseRemoteAddress(channel);
            }
        }
        log.warn("Rpc request id[{}], from remoteAddr[{}] stop process, total wait time in queue is [{}], client timeout setting is [{}].", requestCommand.getId(), remoteAddr, (currentTimestamp - requestCommand.getArriveTime()), requestCommand.getTimeout());
    }


}
